package org.example.day20250105.flink.doris;

import com.alibaba.fastjson2.JSON;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.serializer.SimpleStringSerializer;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.day20250105.ExcelDataHandle;
import org.example.day20250105.domain.OdsTestFlow;
import org.example.day20250105.flink.DorisConstant;

import java.util.Properties;

/**
 * @className: OdsTestFlowWriteByDoris
 * @author: 孙磊
 * @date: 2025/1/11 20:02
 * @Version: 1.0
 * @description: datastream写入到
 */
public class OdsTestFlowWriteByDoris {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setParallelism(1);
        DorisSink.Builder<String> builder = DorisSink.builder();
        DorisOptions.Builder dorisOptions = DorisOptions.builder();
        DorisOptions options = dorisOptions.setFenodes(DorisConstant.FE_NODE_URL)
                .setTableIdentifier(DorisConstant.DB_NAMME + ".ods_xw_test_flow")
                .setUsername(DorisConstant.USERNAME)
                .setPassword(DorisConstant.PASSWORD).build();
        Properties properties = new Properties();
        // 上游是 json 写入时，需要开启配置
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");
        DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
        executionBuilder.setLabelPrefix("label-doris") //streamload label prefix
                .setDeletable(false)
                .setStreamLoadProp(properties);

        builder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(executionBuilder.build())
                .setSerializer(new SimpleStringSerializer()) //serialize according to string
                .setDorisOptions(options);

        env.fromData(ExcelDataHandle.readTestFlowData())
                .map(new MapFunction<OdsTestFlow, String>() {
                    @Override
                    public String map(OdsTestFlow odsTestFlow) throws Exception {
                        return JSON.toJSONString(odsTestFlow);
                    }
                })
                .sinkTo(builder.build());

        env.execute();
    }
}
